package cn.metona.mq.core;

import cn.metona.mq.consumer.MessageConsumer;
import cn.metona.mq.config.ConfigManager;
import cn.metona.mq.exception.MessageRejectedException;
import cn.metona.mq.utils.DLQManager;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

/**
 * 消息队列核心实现
 * 支持优先级、延迟消息、持久化等功能
 */
public class MessageQueue {
    private static final Logger logger = Logger.getLogger(MessageQueue.class.getName());

    private final String queueName;                           // 队列名称
    private final CustomPriorityQueue queue;                  // 自定义队列
    private final MessageStore messageStore;                  // 消息存储
    private final AtomicLong messageCount;                    // 消息计数器
    private final ScheduledExecutorService scheduler;         // 定时任务执行器
    private final List<MessageConsumer> consumers;            // 消费者列表
    private final AtomicBoolean running;                      // 队列运行状态（原子操作）
    private final AtomicBoolean paused;                       // 队列暂停状态（原子操作）
    private final DLQManager dlqManager;                      // 死信队列管理器
    private final ConfigManager configManager;                // 配置管理器
    private final AtomicLong processedCount;                  // 已处理消息计数
    private final AtomicLong failedCount;                     // 失败消息计数

    /**
     * 自定义队列实现
     */
    private static class CustomPriorityQueue {
        private final java.util.PriorityQueue<QueueMessageWrapper> internalQueue;
        private final Object lock = new Object();

        public CustomPriorityQueue() {
            this.internalQueue = new java.util.PriorityQueue<>(11, QueueMessageWrapper::compareTo);
        }

        public void offer(QueueMessageWrapper message) {
            synchronized (lock) {
                internalQueue.offer(message);
                lock.notifyAll();
            }
        }

        public QueueMessageWrapper take(AtomicBoolean paused) throws InterruptedException {
            synchronized (lock) {
                while (internalQueue.isEmpty() || paused.get()) {
                    if (paused.get()) {
                        // 队列暂停时抛出特定异常
                        throw new QueuePausedException("Queue is paused");
                    }
                    lock.wait(100); // 短时间等待
                }

                // 检查消息是否到期
                QueueMessageWrapper first = internalQueue.peek();
                if (first != null) {
                    long now = System.nanoTime();
                    long scheduledTime = first.getScheduledTimeNanos();

                    if (now >= scheduledTime) {
                        // 消息已到期，可以消费
                        return internalQueue.poll();
                    } else {
                        // 消息未到期，等待一小段时间
                        long waitTimeMillis = Math.min((scheduledTime - now) / 1_000_000, 100);
                        if (waitTimeMillis > 0) {
                            lock.wait(waitTimeMillis);
                        }
                        // 重新检查
                        if (!internalQueue.isEmpty() && System.nanoTime() >= internalQueue.peek().getScheduledTimeNanos()) {
                            return internalQueue.poll();
                        }
                        throw new InterruptedException("Message not ready");
                    }
                }

                throw new InterruptedException("No message available");
            }
        }

        public QueueMessageWrapper poll(AtomicBoolean paused) {
            synchronized (lock) {
                if (internalQueue.isEmpty() || paused.get()) {
                    return null;
                }

                QueueMessageWrapper first = internalQueue.peek();
                if (first != null && System.nanoTime() >= first.getScheduledTimeNanos()) {
                    return internalQueue.poll();
                }
                return null;
            }
        }

        public void clear() {
            synchronized (lock) {
                internalQueue.clear();
                lock.notifyAll();
            }
        }

        public int size() {
            synchronized (lock) {
                return internalQueue.size();
            }
        }

        public void notifyQueueStateChange() {
            synchronized (lock) {
                lock.notifyAll();
            }
        }
    }

    /**
     * 队列暂停异常
     */
    public static class QueuePausedException extends InterruptedException {
        public QueuePausedException(String message) {
            super(message);
        }
    }

    /**
     * 队列消息包装器
     */
    private static class QueueMessageWrapper implements Comparable<QueueMessageWrapper> {
        private final QueueMessage queueMessage;
        private final long scheduledTimeNanos;
        private final long createTimeNanos;

        public QueueMessageWrapper(QueueMessage queueMessage) {
            this.queueMessage = queueMessage;
            Message message = queueMessage.getMessage();
            if (message.delayTime() > 0) {
                this.scheduledTimeNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(message.delayTime());
            } else {
                this.scheduledTimeNanos = System.nanoTime();
            }
            this.createTimeNanos = System.nanoTime();
        }

        public QueueMessage getQueueMessage() {
            return queueMessage;
        }

        public long getScheduledTimeNanos() {
            return scheduledTimeNanos;
        }

        @Override
        public int compareTo(QueueMessageWrapper other) {
            // 1. 按调度时间排序
            long timeDiff = this.scheduledTimeNanos - other.scheduledTimeNanos;
            if (Math.abs(timeDiff) > 10_000_000) { // 10毫秒
                return timeDiff < 0 ? -1 : 1;
            }

            // 2. 按优先级排序
            int priorityCompare = Integer.compare(
                    other.queueMessage.getMessage().priority(),
                    this.queueMessage.getMessage().priority()
            );
            if (priorityCompare != 0) {
                return priorityCompare;
            }

            // 3. 按创建时间排序
            return Long.compare(this.createTimeNanos, other.createTimeNanos);
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj) return true;
            if (obj == null || getClass() != obj.getClass()) return false;
            QueueMessageWrapper that = (QueueMessageWrapper) obj;
            return queueMessage.getMessageId().equals(that.queueMessage.getMessageId());
        }

        @Override
        public int hashCode() {
            return queueMessage.getMessageId().hashCode();
        }
    }

    /**
     * 构造函数
     */
    public MessageQueue(String queueName) {
        this(queueName, true);
    }

    /**
     * 构造函数 - 可选择是否启用持久化
     */
    public MessageQueue(String queueName, boolean enablePersistence) {
        this.queueName = queueName;
        this.queue = new CustomPriorityQueue();

        this.messageStore = new MessageStore(queueName, enablePersistence);
        this.messageCount = new AtomicLong(0);
        this.scheduler = Executors.newScheduledThreadPool(2, r -> {
            Thread t = new Thread(r, "MessageQueue-Scheduler-" + queueName);
            t.setDaemon(true);
            return t;
        });
        this.consumers = new CopyOnWriteArrayList<>();
        this.running = new AtomicBoolean(true);
        this.paused = new AtomicBoolean(false);
        this.dlqManager = new DLQManager();
        this.configManager = ConfigManager.getInstance();
        this.processedCount = new AtomicLong(0);
        this.failedCount = new AtomicLong(0);

        // 恢复持久化消息
        if (enablePersistence) {
            messageStore.recoverMessages();
            messageCount.set(messageStore.getMessageCount());
        }

        // 启动监控任务
        startMonitorTask();
    }

    /**
     * 发送消息到队列
     */
    public void sendMessage(Message message) {
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }

        if (!running.get()) {
            throw new IllegalStateException("Queue is not running: " + queueName);
        }

        // 检查队列容量
        int maxCapacity = configManager.getIntProperty("mq.default.queue.capacity", 10000);
        if (messageCount.get() >= maxCapacity) {
            throw new MessageRejectedException("Queue is full: " + queueName);
        }

        QueueMessage queueMessage = new QueueMessage(message);

        // 检查消息是否已经存在（去重）
        if (messageStore.getMessage(message.messageId()) != null) {
            logger.warning("Duplicate message detected: " + message.messageId());
            return;
        }

        QueueMessageWrapper wrapper = new QueueMessageWrapper(queueMessage);
        queue.offer(wrapper);
        messageStore.storeMessage(queueMessage);
        messageCount.incrementAndGet();

        logger.fine("Message sent to queue: " + queueName + ", messageId: " + message.messageId());
    }

    /**
     * 批量发送消息
     */
    public void sendMessages(List<Message> messages) {
        if (messages == null) {
            throw new IllegalArgumentException("Messages list cannot be null");
        }

        for (Message message : messages) {
            sendMessage(message);
        }
    }

    /**
     * 消费消息
     */
    public QueueMessage consumeMessage() throws InterruptedException {
        if (!running.get()) {
            throw new IllegalStateException("Queue is not running: " + queueName);
        }

        try {
            QueueMessageWrapper wrapper = queue.take(paused);
            QueueMessage message = wrapper.getQueueMessage();
            message.setConsumed(true);
            logger.fine("Message consumed from queue: " + queueName + ", messageId: " + message.getMessageId());
            return message;
        } catch (QueuePausedException e) {
            logger.fine("Queue is paused, consumer will wait: " + queueName);
            throw e;
        }
    }

    /**
     * 异步消费消息（非阻塞）
     */
    public QueueMessage consumeMessageAsync() {
        if (!running.get()) {
            return null;
        }

        QueueMessageWrapper wrapper = queue.poll(paused);
        if (wrapper != null) {
            QueueMessage message = wrapper.getQueueMessage();
            message.setConsumed(true);
            logger.fine("Message consumed async from queue: " + queueName + ", messageId: " + message.getMessageId());
            return message;
        }
        return null;
    }

    /**
     * 确认消息消费
     */
    public boolean ackMessage(String messageId) {
        if (messageId == null || messageId.isEmpty()) {
            return false;
        }

        QueueMessage message = messageStore.getMessage(messageId);
        if (message != null) {
            message.setAcked(true);
            messageStore.removeMessage(messageId);
            messageCount.decrementAndGet();
            processedCount.incrementAndGet();
            logger.fine("Message acknowledged: " + messageId);
            return true;
        }
        return false;
    }

    /**
     * 消息消费失败，重新入队
     */
    public boolean nackMessage(String messageId) {
        if (messageId == null || messageId.isEmpty()) {
            return false;
        }

        QueueMessage message = messageStore.getMessage(messageId);
        if (message != null) {
            message.incrementRetryCount();
            message.resetConsumeState();

            int maxRetries = configManager.getIntProperty("mq.default.retry.count", 3);
            if (message.shouldRetry(maxRetries)) {
                QueueMessageWrapper wrapper = new QueueMessageWrapper(message);
                queue.offer(wrapper);
                logger.info("Message nack and requeued: " + messageId +
                        ", retry count: " + message.getRetryCount());
            } else {
                moveToDLQ(message);
            }
            return true;
        }
        return false;
    }

    /**
     * 移动消息到死信队列
     */
    private void moveToDLQ(QueueMessage message) {
        try {
            dlqManager.moveToDLQ(queueName, message.getMessage());
            messageStore.removeMessage(message.getMessageId());
            messageCount.decrementAndGet();
            failedCount.incrementAndGet();
            logger.severe("Message moved to DLQ: " + message.getMessageId());
        } catch (Exception e) {
            logger.severe("Failed to move message to DLQ: " + message.getMessageId() +
                    ", error: " + e.getMessage());
        }
    }

    /**
     * 获取队列大小
     */
    public long size() {
        return messageCount.get();
    }

    /**
     * 获取队列名称
     */
    public String getQueueName() {
        return queueName;
    }

    /**
     * 添加消费者
     */
    public void addConsumer(MessageConsumer consumer) {
        if (consumer != null) {
            consumers.add(consumer);
        }
    }

    /**
     * 移除消费者
     */
    public void removeConsumer(MessageConsumer consumer) {
        if (consumer != null) {
            consumers.remove(consumer);
        }
    }

    /**
     * 获取消费者数量
     */
    public int getConsumerCount() {
        return consumers.size();
    }

    /**
     * 启动队列
     */
    public void start() {
        boolean wasRunning = running.getAndSet(true);
        if (!wasRunning) {
            logger.info("Queue started: " + queueName);
            queue.notifyQueueStateChange(); // 通知所有等待的线程
        }
    }

    /**
     * 停止队列
     */
    public void stop() {
        boolean wasRunning = running.getAndSet(false);
        if (wasRunning) {
            paused.set(false);
            scheduler.shutdown();
            messageStore.close();
            logger.info("Queue stopped: " + queueName);
            queue.notifyQueueStateChange(); // 通知所有等待的线程
        }
    }

    /**
     * 暂停队列
     */
    public void pause() {
        boolean wasPaused = paused.getAndSet(true);
        if (!wasPaused) {
            logger.info("Queue paused: " + queueName);
            queue.notifyQueueStateChange(); // 通知所有等待的线程
        }
    }

    /**
     * 恢复队列
     */
    public void resume() {
        boolean wasPaused = paused.getAndSet(false);
        if (wasPaused) {
            logger.info("Queue resumed: " + queueName);
            queue.notifyQueueStateChange(); // 通知所有等待的线程
        }
    }

    /**
     * 清空队列
     */
    public void clear() {
        queue.clear();
        messageStore.clear();
        messageCount.set(0);
        logger.info("Queue cleared: " + queueName);
    }

    /**
     * 启动监控任务
     */
    private void startMonitorTask() {
        scheduler.scheduleAtFixedRate(() -> {
            if (!running.get()) return;

            logger.fine("Queue monitor - Name: " + queueName +
                    ", Size: " + size() +
                    ", Consumers: " + getConsumerCount() +
                    ", Processed: " + processedCount.get() +
                    ", Failed: " + failedCount.get() +
                    ", Paused: " + paused.get() +
                    ", Running: " + running.get());
        }, 30, 30, TimeUnit.SECONDS);
    }

    /**
     * 获取队列统计信息
     */
    public QueueStats getStats() {
        return new QueueStats(queueName, size(), getConsumerCount(), running.get(),
                processedCount.get(), failedCount.get(), paused.get());
    }

    /**
     * 获取消息存储
     */
    public MessageStore getMessageStore() {
        return messageStore;
    }

    /**
     * 检查队列是否暂停
     */
    public boolean isPaused() {
        return paused.get();
    }

    /**
     * 检查队列是否运行
     */
    public boolean isRunning() {
        return running.get();
    }

    /**
     * 队列统计信息内部类
     */
    public static class QueueStats {
        private final String queueName;
        private final long messageCount;
        private final int consumerCount;
        private final boolean running;
        private final long processedCount;
        private final long failedCount;
        private final boolean paused;

        public QueueStats(String queueName, long messageCount, int consumerCount,
                          boolean running, long processedCount, long failedCount, boolean paused) {
            this.queueName = queueName;
            this.messageCount = messageCount;
            this.consumerCount = consumerCount;
            this.running = running;
            this.processedCount = processedCount;
            this.failedCount = failedCount;
            this.paused = paused;
        }

        // Getter方法
        public String getQueueName() { return queueName; }
        public long getMessageCount() { return messageCount; }
        public int getConsumerCount() { return consumerCount; }
        public boolean isRunning() { return running; }
        public long getProcessedCount() { return processedCount; }
        public long getFailedCount() { return failedCount; }
        public boolean isPaused() { return paused; }

        @Override
        public String toString() {
            return String.format("QueueStats{queueName='%s', messageCount=%d, consumerCount=%d, " +
                            "running=%s, processed=%d, failed=%d, paused=%s}",
                    queueName, messageCount, consumerCount, running, processedCount, failedCount, paused);
        }
    }
}